from kafka import *

consumer_path = '../../trans_file/consumer_file/'


def record_msg_piece():
    consumer = KafkaConsumer('testPython',
                             bootstrap_servers=['master:9092'],
                             consumer_timeout_ms=10000)
    msg_dict = {}
    for message in consumer:
        print("%s:%d:%d: key=%s value=%s" %
              (message.topic, message.partition,
               message.offset, message.key,
               message.value[:20]))
        msg_dict[int.from_bytes(message.key,byteorder = 'big')] = message.value
        if message.value == None :
            break
        print(message.value)
        print(1)

    write_file(msg_dict)
    pass

def write_file(msg_dict):
    file_name = 'after_1000.txt'
    with open(consumer_path+file_name, "wb") as w:
        for i in range(len(msg_dict)):
            print("写入文件{}".format(i+1))
            w.write(msg_dict[i+1])

if __name__ == '__main__':
    record_msg_piece()

    pass